package com.hexiang.rocketmq.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

@Component
@Slf4j
public class ComplexBatchMessageMQProducer {

    private String topic = "test_complex_batch_topic";

    private String producerGroup = "complex_batch_producer_group_001";

    private String nameServer = "127.0.0.1:9876";

    private DefaultMQProducer defaultMQProducer;

    @PostConstruct
    public void start() throws MQClientException {
        defaultMQProducer = new DefaultMQProducer(producerGroup, getAclRPCHook());
        defaultMQProducer.setNamesrvAddr(nameServer);
        defaultMQProducer.start();
    }

    @PreDestroy
    public void shutdown() {
        defaultMQProducer.shutdown();
    }

    RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials("rocketmq_u1", "12345678"));
    }

    /**
     * 简单批量发送消息
     */
    public void sendComplexBatchMsg() {

        List<Message> messages = new ArrayList<>();
        for (int i = 0; i < 100; i++) {
            messages.add(new Message(topic, "", "", ("Hello world " + i + "").getBytes()));
        }

        //把大的消息分裂成若干个小的消息
        ListSplitter splitter = new ListSplitter(messages);
        while (splitter.hasNext()) {
            try {
                List<Message> listItem = splitter.next();
                SendResult sendResult = defaultMQProducer.send(listItem);
                log.warn("sendResult " + sendResult);
            } catch (Exception e) {
                log.error("sendComplexBatchMsgErr", e);

            }
        }

    }

    class ListSplitter implements Iterator<List<Message>> {

        private final List<Message> messages;
        private int currIndex;

        public ListSplitter(List<Message> messages) {
            this.messages = messages;
        }

        @Override
        public boolean hasNext() {
            return currIndex < messages.size();
        }

        @Override
        public List<Message> next() {
            int startIndex = getStartIndex();
            int nextIndex = startIndex;
            int totalSize = 0;
            for (; nextIndex < messages.size(); nextIndex++) {
                Message message = messages.get(nextIndex);
                int tmpSize = calcMessageSize(message);
                if (tmpSize + totalSize > defaultMQProducer.getMaxMessageSize()) {
                    break;
                } else {
                    totalSize += tmpSize;
                }
            }
            List<Message> subList = messages.subList(startIndex, nextIndex);
            currIndex = nextIndex;
            return subList;
        }

        private int getStartIndex() {
            Message currMessage = messages.get(currIndex);
            int tmpSize = calcMessageSize(currMessage);
            while (tmpSize > defaultMQProducer.getMaxMessageSize()) {
                currIndex += 1;
                Message message = messages.get(currIndex);
                tmpSize = calcMessageSize(message);
            }
            return currIndex;
        }

        private int calcMessageSize(Message message) {
            int tmpSize = message.getTopic().length() + message.getBody().length;
            Map<String, String> properties = message.getProperties();
            for (Map.Entry<String, String> entry : properties.entrySet()) {
                tmpSize += entry.getKey().length() + entry.getValue().length();
            }
            tmpSize = tmpSize + 20; // 增加⽇日志的开销20字节
            return tmpSize;
        }
    }
}